package com.zhuhao.basic.collection.queue;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.thread.ThreadFactoryBuilder;
import cn.hutool.core.util.StrUtil;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author haozhu7
 * @date 2023/3/11
 */
public class ArrayBlockQueueingDemo {

    /**
     * 用于缓存的数组队列
     */
    private static final BlockingQueue<String> messageQueue = new ArrayBlockingQueue<String>(10240);

    private static volatile boolean isRunning = false;

    /**
     * 批处理的大小
     */
    private static final int HOLDER = 2;

    private final Executor poolExecutor = new ThreadPoolExecutor(2, 3, 10, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1024), ThreadFactoryBuilder.create().setNamePrefix("halk").build());
//    private static final ExecutorService poolExecutor = Executors.newSingleThreadExecutor();


    @Before
    public void init() {
        System.out.println("init");
        isRunning = true;
        poolExecutor.execute(new HandlerTask());
    }

    @After
    public void destroy() {
        System.out.println("destroy");
        isRunning = false;
        if (messageQueue.size() > 0) {
            System.out.println("处理业务");
        }
    }

    public void add(List<String> list) {
        for (String str : list) {
            add(str);
        }
    }

    private void add(String str) {
        try {
            messageQueue.put(str);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private class HandlerTask extends Thread {
        HandlerTask() {
            this.setName("HandlerTask-Handler-Thread");
        }

        @Override
        public void run() {
            List<String> messageCache = new ArrayList<>(HOLDER);
            while (isRunning) {
                System.out.println("任务处理完了，又一个循环处理：" + new Date());
                try {
                    messageCache.clear();
                    while (messageCache.size() < HOLDER) {
                        System.out.println("等待任务进来：" + new Date());
                        String str = messageQueue.poll(1, TimeUnit.SECONDS);
                        if (StrUtil.isBlank(str)) {
                            break;
                        }
                        messageCache.add(str);
                    }
                    if (CollectionUtil.isNotEmpty(messageCache)) {
                        //处理业务
                        businessProcess(messageCache);
                    }
                } catch (Throwable e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 业务逻辑处理
     *
     * @param list
     */
    private void businessProcess(List<String> list) {
        try {
            System.out.println("业务逻辑开始批处理中");
            for (String str : list) {
                System.out.println("处理业务：" + str);
                Thread.sleep(1000);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    @Test
    public void test01() {
        ArrayBlockQueueingDemo queueingDemo = new ArrayBlockQueueingDemo();


        System.out.println("主线程开始。。。");
        System.out.println("缓存队列的数量： " + messageQueue.size());

        //第一批需要处理的
        List<String> firstList = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            firstList.add("first_" + (i + 1));
        }
        queueingDemo.add(firstList);
        System.out.println("缓存队列的数量： " + queueingDemo.messageQueue.size());


        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        //第二批需要处理的
        List<String> secondList = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            secondList.add("second_" + (i + 1));
        }
        queueingDemo.add(secondList);
        System.out.println("缓存队列的数量： " + queueingDemo.messageQueue.size());


        System.out.println("主线程结束。。。");

        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
